package org.comanche.snaf.core.polling;

import java.util.Iterator;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicReference;

import org.comanche.snaf.core.service.Processor;
import org.comanche.snaf.core.session.IoSession;

/**
 * use polling strategy to monitor the read and write queue
 * and monitor the session status
 * 
 * @author joe_zhou
 *
 */
public abstract class AbstractPollingProcess<S extends IoSession> implements Processor<S> {
	
	///////////////////////////////////////////////////
	// private members
	// new session queue
	private Queue<S> newSessions = new ConcurrentLinkedQueue<S>();
	// flush sessions queue
	private Queue<S> flushSessions = new ConcurrentLinkedQueue<S>();
	// remove sessions queue
	private Queue<S> removeSessions = new ConcurrentLinkedQueue<S>();
	private AtomicReference<Processor> processRef = new AtomicReference<Processor>();
	// thread pool
	private Executor excutor;

	private static final int PROCESS_TIMEOUT = 1000;
	
	///////////////////////////////////////////////////
	// inner class
	// this class is the job for handling the newSessions
	// and flushSessions, removeSessions
	private class Processor implements Runnable {
		
		public void run() {
			while(true) {
				// selector in default time out
				try {
					int select = select(PROCESS_TIMEOUT);
					// handle new session
					handleNewSession();
					// process the read/write/disconnect event
					if (select > 0){
						//handleProcess();
						handleSessions(getSessions());
					}
					
					
				} catch (Exception e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				
			}
		}
	}
	
	///////////////////////////////////////////////////
	// private
	private void handleNewSession(){
		
		for(;;){
			S session;
			// get each new session and bind with the listener
			if (newSessions.size() > 0 && (session = newSessions.poll()) != null){
				
				try {
					bindInternal(session);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
			
		}
		
	}
	
	private void handleSessions(Iterator<S> sessions){
		
		while (sessions.hasNext()){
			S s = sessions.next();
			try {
				// is readable do read process
				// is writeable do write process
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}
	
	///////////////////////////////////////////////////
	// protected field
	// initialize the field
	protected abstract void init() throws Exception;
	// select the wait by the time out
	// if the time out is 0 or less than 0, then wait until wait up is called
	// or some channel is active
	protected abstract int select(int timeout) throws Exception;
	// wake up function
	protected abstract void wakeup();
	// bind the register
	protected abstract void bindInternal(S session) throws Exception;
	// get sessions
	protected abstract Iterator<S> getSessions();
	
	///////////////////////////////////////////////////
	// interface implements
	public void add(S session) {
		if (session != null){
			newSessions.add(session);
			// begin to monitor when the first time
			if (processRef.get() == null){
				Processor processor = new Processor();
				if (processRef.compareAndSet(null, processor)){
					excutor.execute(processor);
				}
			}
		}
	}
	
	public void remove(S session) {
		removeSessions.remove(session);
	}
	
	public void flush(S session) {
		flushSessions.add(session);
	}	

}
